package m3;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Scanner;

public class Producer {
    public static void main(String[] args) throws MQClientException {
        //创建事务消息生产者
        TransactionMQProducer p = new TransactionMQProducer("p3");
        //设置 name server
        p.setNamesrvAddr("192.168.64.141:9876");
        //设置事务消息的监听器，监听消息发送
        p.setTransactionListener(new TransactionListener() {
            //执行本地事务
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                //模拟网络中断，无法通知服务器本地事务的执行状态
                if (Math.random() < 1) {
                    System.out.println("模拟网络中断，无法通知rocketmq本地事务的执行状态");
                    return LocalTransactionState.UNKNOW;
                }

                System.out.println("执行本地事务，参数："+arg);

                if (Math.random() < 0.5) {
                    System.out.println("成功，提交消息");
                    return LocalTransactionState.COMMIT_MESSAGE;    //成功，提交消息
                } else {
                    System.out.println("失败，回滚消息");
                    return LocalTransactionState.ROLLBACK_MESSAGE;  //失败，撤回消息
                }
            }
            //处理rocketmq服务器的反向回查
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                System.out.println("rocketmq正在回查事务状态");
                return LocalTransactionState.UNKNOW;
            }
        });
        //启动
        p.start();
        //发送事务消息（半消息）,会触发一个监听器来执行本地事务
        while (true) {
            System.out.print("输入消息： ");
            String s = new Scanner(System.in).nextLine();
            Message msg = new Message("Topic3", s.getBytes());
            p.sendMessageInTransaction(msg, "传递到监听器的业务数据参数");
        }
    }
}
